import xml.etree.cElementTree as ET
import pprint
filename = "C:/Users/oikonomakisa/Desktop/example.osm"
def count_tags(filename):
tags = {}
for event, elem in ET.iterparse(filename):
if elem.tag in tags:
tags[elem.tag] += 1
else:
tags[elem.tag] = 1
print tags
return tags
#def test():
# tags = count_tags('example.osm')
# pprint.pprint(tags)
# assert tags == {'bounds': 1,
# 'member': 3,
# 'nd': 4,
# 'node': 20,
# 'osm': 1,
# 'relation': 1,
# 'tag': 7,
# 'way': 1}
if __name__ == "__main__":
count_tags(filename)
import xml.etree.cElementTree as ET
import pprint
import re
filename = "C:/Users/oikonomakisa/Desktop/example2.osm"
lower = re.compile(r'^([a-z]|_)*$')
lower_colon = re.compile(r'^([a-z]|_)*:([a-z]|_)*$')
problemchars = re.compile(r'[=\+/&<>;\'"\?%#$@\,\. \t\r\n]')
def key_type(element, keys):
if element.tag == "tag":
k = element.get("k")
if re.search(lower,k):
keys["lower"] += 1
elif re.search(lower_colon,k):
keys["lower_colon"] += 1
elif re.search(problemchars,k):
keys["problemchars"] +=1
else:
keys["other"] += 1
return keys
def process_map(filename):
keys = {"lower": 0, "lower_colon": 0, "problemchars": 0, "other": 0}
for _, element in ET.iterparse(filename):
keys = key_type(element, keys)
pprint.pprint(keys)
return keys
if __name__ == "__main__":
process_map(filename)
import xml.etree.cElementTree as ET
import pprint
import re
filename = "C:/Users/oikonomakisa/Desktop/example3.osm"
def get_user(element,users):
if 'uid' in element.attrib.keys():
user = element.attrib['uid']
users.add(user)
pass
return users
def process_map(filename):
users = set()
for _, element in ET.iterparse(filename):
users = get_user(element, users)
pass
pprint.pprint(users)
return users
if __name__ == "__main__":
process_map(filename)
import xml.etree.cElementTree as ET
from collections import defaultdict
import re
import pprint
OSMFILE = "C:/Users/oikonomakisa/Desktop/example4.osm"
street_type_re = re.compile(r'\b\S+\.?$', re.IGNORECASE)
expected = ["Street", "Avenue", "Boulevard", "Drive", "Court", "Place", "Square", "Lane", "Road",
"Trail", "Parkway", "Commons"]
mapping = { "St": "Street",
"St.": "Street",
"Ave": "Avenue",
"Rd.": "Road"
}
def audit_street_type(street_types, street_name):
m = street_type_re.search(street_name)
if m:
street_type = m.group()
if street_type not in expected:
street_types[street_type].add(street_name)
def is_street_name(elem):
return (elem.attrib['k'] == "addr:street")
def audit(osmfile):
osm_file = open(osmfile, "r")
street_types = defaultdict(set)
for event, elem in ET.iterparse(osm_file, events=("start",)):
if elem.tag == "node" or elem.tag == "way":
for tag in elem.iter("tag"):
if is_street_name(tag):
audit_street_type(street_types, tag.attrib['v'])
return street_types
def update_name(name, mapping):
m = street_type_re.search(name)
if m:
street_type = m.group()
if street_type not in expected:
name = re.sub(street_type_re, mapping[street_type], name)
return name
def test():
st_types = audit(OSMFILE)
assert len(st_types) == 3
pprint.pprint(dict(st_types))
for st_type, ways in st_types.iteritems():
for name in ways:
better_name = update_name(name, mapping)
print name, "=>", better_name
if name == "West Lexington St.":
assert better_name == "West Lexington Street"
if name == "Baldwin Rd.":
assert better_name == "Baldwin Road"
if __name__ == '__main__':
test()
import xml.etree.cElementTree as ET
import pprint
import re
import codecs
import json
OSMFILE = "C:/Users/oikonomakisa/Desktop/example5.osm"
lower = re.compile(r'^([a-z]|_)*$')
lower_colon = re.compile(r'^([a-z]|_)*:([a-z]|_)*$')
problemchars = re.compile(r'[=\+/&<>;\'"\?%#$@\,\. \t\r\n]')
CREATED = [ "version", "changeset", "timestamp", "user", "uid"]
def shape_element(element):
node = {}
if element.tag == "node" or element.tag == "way" :
node["created"] = {CREATED[0]: element.attrib["version"],
CREATED[1]: element.attrib["changeset"],
CREATED[2]: element.attrib["timestamp"],
CREATED[3]: element.attrib["user"],
CREATED[4]: element.attrib["uid"]}
is_addresspart = re.compile(r'^addr:([a-z]|_)*$')
is_street_part = re.compile(r'^addr:([a-z]|_)*:([a-z]|_)*$')
this_address = {}
flag = False
for tag in element.iter("tag"):
if re.search(problemchars, tag.attrib["k"]):
break
elif re.search(is_street_part, tag.attrib["k"]):
break
elif re.search(is_addresspart, tag.attrib["k"]):
this_address[tag.attrib["k"][5:]] = tag.attrib["v"]
flag = True
elif tag.attrib["k"] == "cuisine":
node["cuisine"] = tag.attrib["v"]
elif tag.attrib["k"] == "amenity":
node["amenity"] = tag.attrib["v"]
elif tag.attrib["k"] == "name":
node["name"] = tag.attrib["v"]
elif tag.attrib["k"] == "phone":
node["phone"] = tag.attrib["v"]
else:
break
if flag == True:
node["address"] = this_address
node_ref = []
flag = False
for tag in element.iter("nd"):
node_ref.append(tag.attrib["ref"])
flag = True
if flag == True:
node["node_refs"] = node_ref
try:
node["pos"] = [float(element.attrib["lat"]), float(element.attrib["lon"])]
except:
pass
node["type"] = element.tag
try:
node["visible"] = element.attrib["visible"]
except:
pass
node["id"] = element.attrib["id"]
pprint.pprint(node)
return node
else:
return None
def process_map(OSMFILE, pretty = False):
# You do not need to change this file
file_out = "{0}.json".format(OSMFILE)
data = []
with codecs.open(file_out, "w") as fo:
for _, element in ET.iterparse(OSMFILE):
el = shape_element(element)
if el:
data.append(el)
if pretty:
fo.write(json.dumps(el, indent=2)+"\n")
else:
fo.write(json.dumps(el) + "\n")
return data
def test():
# NOTE: if you are running this code on your computer, with a larger dataset,
# call the process_map procedure with pretty=False. The pretty=True option adds
# additional spaces to the output, making it significantly larger.
data = process_map(OSMFILE, False)
#pprint.pprint(data)
correct_first_elem = {
"id": "261114295",
"visible": "true",
"type": "node",
"pos": [41.9730791, -87.6866303],
"created": {
"changeset": "11129782",
"user": "bbmiller",
"version": "7",
"uid": "451048",
"timestamp": "2012-03-28T18:31:23Z"
}
}
assert data[0] == correct_first_elem
assert data[-1]["address"] == {
"street": "West Lexington St.",
"housenumber": "1412"
}
assert data[-1]["node_refs"] == [ "2199822281", "2199822390", "2199822392", "2199822369",
"2199822370", "2199822284", "2199822281"]
if __name__ == "__main__":
test()
import xml.etree.cElementTree as ET
import codecs
import pprint
import json
import re
import os
os.getcwd()
os.chdir('C:/Users/oikonomakisa/Desktop/')
lower = re.compile(r'^([a-z]|_)*$')
lower_colon = re.compile(r'^([a-z]|_)*:([a-z]|_)*$')
problemchars = re.compile(r'[=\+/&<>;\'"\?%#$@\,\. \t\r\n]')
street_type_re = re.compile(r'\b\S+\.?$', re.IGNORECASE)
mapping = { "St": "Street",
"St.": "Street",
"Ave": "Avenue",
"Rd.": "Road",
"avenue": "Avenue",
"street": "Street"
}
CREATED = ["version", "changeset", "timestamp", "user", "uid"]
def get_pos(element):
"""Returns the latitude and longitude of the element in an array."""
lat = float(element.attrib['lat'])
lon = float(element.attrib['lon'])
pos = [lat, lon]
return pos
def ignoring(k):
"""Returns True if key k should be ignored."""
KEYS = ["Street", "Avenue", "Boulevard", "Parade", "Wakefield/Cuba", "Place", "Square", "Lane", "Road", "Terrace", "Crescent", "Way", "Grove",
"Trail", "Parkway", "Commons", "Drive", "Esplanade", "Quebec", "South"]
if k in KEYS:
return True
return False
def postcode_checker(v):
"""
Checks postcodes and reduces to 4 digit strings.
"""
postcode = ''
for char in v:
if char.isdigit():
postcode += char
if len(postcode) == 4:
break
return postcode
def node_update_k(node, value, tag):
"""Adds 'k' and 'v' values from tag as new key:value pair to node."""
k = value
v = tag.attrib['v']
if k.startswith('addr:'):
# Ignore 'addr:street:' keys with 2 colons
if k.count(':') == 1:
if 'address' not in node:
node['address'] = {}
if k == 'addr:postcode':
node['address']['postcode'] = postcode_checker(v)
elif k == 'addr:street':
node['address'][k[5:]] = v
# Process other k:v pairs normally
else:
node[k] = v
return node
def shape_element(element):
"""
Takes an XML tag as input and returns a cleaned and reshaped
dictionary for JSON ouput. If the element contains an abbreviated
street name, it returns with an updated full street name.
"""
node = {}
if element.tag == "node" or element.tag == "way" :
node['type'] = element.tag
node['created'] = {}
if 'lat' in element.attrib:
# Get coordinates
node['pos'] = get_pos(element)
# Begin iterating over subtags
for tag in element.iter():
for key, value in tag.items():
if key in CREATED:
node['created'][key] = value
# Check for problem characters and ignored values
# in second-level tag 'k' attributes
elif key == 'k' and not re.search(problemchars, value):
if not ignoring(value):
node = node_update_k(node, value, tag)
# Create/update array 'node_refs'
elif key == 'ref':
if 'node_refs' not in node:
node['node_refs'] = []
node['node_refs'].append(value)
# Process remaining tags
elif key == "cuisine":
node["cuisine"] = value
elif key == "amenity":
node["amenity"] = value
elif key == "name":
node["name"] = value
elif key == "phone":
node["phone"] = value
# Safe to clear() now that element has been processed
element.clear()
return node
else:
return None
def process_map(file_in, pretty = False):
"""
Outputs a JSON file with the above structure.
Returns the data as a list of dictionaries.
If running main_test(), comment out all array 'data' operations.
"""
file_out = "{0}.json".format(file_in)
data = []
with codecs.open(file_out, "w") as fo:
parser = ET.iterparse(file_in)
for __, elem in parser:
el = shape_element(elem)
if el:
data.append(el)
# Output to JSON
if pretty:
fo.write(json.dumps(el, indent=2)+"\n")
else:
fo.write(json.dumps(el) + "\n")
del parser
return data
def main_test():
data = process_map('wellington_sample.osm', False)
print len(data)
print 'Map processed'
if __name__ == '__main__':
main_test()